package com.project.pulsar.delayMsgAndSubscriptionsModel;

import com.project.pulsar.conf.PulsarConf;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.TimeUnit;

/**
 * 延时消息
 */
@RestController
public class DelayMessageController {
    @Autowired
    PulsarConf pulsarConf;

    String tenantName = "delay";//调用接口创建的租户,创建方法见com.project.pulsar.persistent.PersistentController.java
    String namespace = "mySpace";//调用接口创建的命名空间,创建方法见com.project.pulsar.persistent.PersistentController.java

    /**
     * 生产消息
     *
     * @param msg
     * @throws PulsarClientException
     */
    @GetMapping("/delay/sendMsg")
    public void sendMsg(String msg) throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        String topic = "persistent://" + tenantName + "/" + namespace + "/my-delay-topic";
        //持久化         租户   命名空间   主题
        Producer<byte[]> producer = pulsarFactory.newProducer()
                .topic(topic)
                .create();
        producer.newMessage().deliverAfter(2L, TimeUnit.MINUTES).key("key-1").value("message-1-1".getBytes()).send();
        //        //    MessageId send = producer1.newMessage().deliverAfter(3L, TimeUnit.MINUTES).value("Hello Pulsar!".getBytes()).send();//指定消息在X时分秒后执行
//        MessageId send =producer1.newMessage().deliverAt(
//                LocalDateTime.of(2021, 9, 20, 16, 20).toInstant(ZoneOffset.of("+8")).toEpochMilli()
//        ).value(msg.getBytes()).send();//指定消息在2021/09/20 15:52分发送，如发送时日期已过，会立刻发送
    }


    /**
     * 手动执行获取消息
     *
     * @throws PulsarClientException
     */
    @GetMapping("/delay/comsumer")
    public void comsumerByArtificial() throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        String topic = "persistent://" + tenantName + "/" + namespace + "/my-delay-topic";
        //持久化         租户   命名空间   主题
        Consumer<byte[]> consumer = pulsarFactory.newConsumer()
                .topic(topic)
                .subscriptionName("my-subscription")//必须名称一致，才可触发下一行，否则就是两个不同的消费者
                .subscriptionType(SubscriptionType.Key_Shared)//指定模式
                .subscribe();
        Message<byte[]> receive = consumer.receive();
        System.out.println(new String(receive.getData()));
        consumer.acknowledge(receive);//确认消息被消费
        consumer.close();
    }

    /**
     * 手动执行获取消息
     *
     * @throws PulsarClientException
     */
    @GetMapping("/delay/comsumer2")
    public void comsumerByArtificial2() throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        String topic = "persistent://" + tenantName + "/" + namespace + "/my-delay-topic";
        //持久化         租户   命名空间   主题
        Consumer<byte[]> consumer = pulsarFactory.newConsumer()
                .topic(topic)
                .subscriptionName("my-subscription")//必须名称一致，才可触发下一行，否则就是两个不同的消费者
                .subscriptionType(SubscriptionType.Key_Shared)//指定模式
                .subscribe();
        Message<byte[]> receive = consumer.receive();
        System.out.println(new String(receive.getData()));
        consumer.acknowledge(receive);//确认消息被消费
        consumer.close();
    }
}